package com.example;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * bin/spark-submit --class com.example.JavaWordCount \
 *     ./spark-0.0.1-SNAPSHOT.jar  \
 *     hdfs://namenode:9000/test/test.txt \
 * 	   hdfs://namenode:9000/output
 */
public class JavaWordCount {

    public static void main(String args[]) {
        SparkConf conf = new SparkConf();
        conf.setAppName("Spark-JavaWordCount");
        //conf.setMaster("local[*]");
        //conf.setMaster("spark://namenode:7077");
        conf.setMaster("yarn");
        conf.set("spark.submit.deployMode", "client");
        // --driver-memory
        conf.set("spark.driver.memory", "1024m");
        // --executor-memory
        conf.set("spark.executor.memory", "1024m");
        // --executor-cores
        conf.set("spark.executor.cores", "1");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> linesRDD = jsc.textFile(args[0]);
        JavaRDD<String> wordsRDD = linesRDD.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
        JavaPairRDD<String, Integer> paresRDD = wordsRDD.mapToPair(x -> new Tuple2<>(x, 1));
        JavaPairRDD<String, Integer> wordCountsRDD = paresRDD.reduceByKey((x, y) -> x + y);
        wordCountsRDD.saveAsTextFile(args[1]);
    }

}
